# data_pipeline.py
import requests
import pandas as pd
import sqlite3
import logging

# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


def fetch_movies(api_key):
    """从TMDB API获取热门电影数据"""
    try:
        url = f"https://api.themoviedb.org/3/movie/popular?api_key={api_key}&language=zh-CN&page=1"
        response = requests.get(url, timeout=15)
        response.raise_for_status()
        data = response.json()

        if 'results' not in data:
            logger.error("API返回数据格式异常：缺少'results'字段")
            return pd.DataFrame()

        logger.info(f"成功从TMDB API获取到 {len(data['results'])} 部电影数据")
        return pd.DataFrame(data['results'])

    except requests.exceptions.RequestException as e:
        logger.error(f"网络请求出错: {e}")
        return pd.DataFrame()
    except Exception as e:
        logger.error(f"获取数据时发生未知错误: {e}")
        return pd.DataFrame()


def create_mock_data():
    """创建模拟电影数据用于演示"""
    import numpy as np
    logger.info("正在生成模拟数据...")

    mock_data = {
        'id': range(1, 21),
        'title': [f'电影《测试示例{i}》' for i in range(1, 21)],
        'release_date': [f'2023-{month:02d}-15' for month in range(1, 13)] +
                        [f'2024-{month:02d}-15' for month in range(1, 9)],
        'popularity': np.random.uniform(20, 150, 20),
        'vote_average': np.round(np.random.uniform(5.0, 8.9, 20), 1),
        'vote_count': np.random.randint(500, 10000, 20),
        'overview': ['这是一部精彩的电影剧情描述'] * 20
    }
    return pd.DataFrame(mock_data)


def process_data(df):
    """处理电影数据"""
    if df.empty:
        return df

    # 处理缺失值
    if 'release_date' in df.columns:
        df['release_date'].fillna('未知', inplace=True)

    logger.info(f"数据处理完成，共 {len(df)} 条记录")
    return df


def save_to_db(df, db_name='movies.db'):
    """保存数据到SQLite数据库"""
    if df.empty:
        logger.warning("尝试保存空数据到数据库")
        return

    try:
        conn = sqlite3.connect(db_name)
        # 选择需要保存到数据库的列
        columns_to_save = ['title', 'release_date', 'popularity', 'vote_average', 'vote_count']
        available_columns = [col for col in columns_to_save if col in df.columns]

        df_to_save = df[available_columns]
        df_to_save.to_sql('movies', conn, if_exists='replace', index=False)
        conn.close()
        logger.info(f"成功保存 {len(df_to_save)} 条数据到数据库")
    except Exception as e:
        logger.error(f"保存到数据库时出错: {e}")